package yuan.producer.intepceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class ProducerInterceptorPrefix implements ProducerInterceptor<String,String> {
    private volatile int onSucccess = 0;
    private volatile int onfailure = 0;

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {

        String prefix = "prefix-"+producerRecord.value();


        return new ProducerRecord<>(
                producerRecord.topic(), producerRecord.partition(), producerRecord.timestamp(), producerRecord.key(), prefix, producerRecord.headers()
        );

    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

        if (e == null) {
            onSucccess++;
        }else{
            onfailure++;

        }

    }

    @Override
    public void close() {
        System.out.printf("%f%",100*onSucccess/(onSucccess+onfailure));
    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
